/*-------------------------------------------------------------------------
 *
 * tablespace.c
 *	  操作表空间的命令
 *
 * PostgreSQL 中的表空间旨在允许用户确定给定数据库对象的数据文件所在的文件系统位置。
 *
 * 表空间表示文件系统上的一个目录。在创建表空间时，目录必须是空的。为了简化事情并消除文件名冲突的可能性，我们将文件隔离在特定于数据库的子目录中。
 *
 * 为了支持通过 RelFileNode 提供的信息进行文件访问，我们在 $PGDATA/pg_tblspc 中维护一个符号链接映射。符号链接按表空间 OID 命名并指向实际的表空间目录。每个表空间中还有一个每集群版本的目录。因此，任意文件的完整路径是
 *			$PGDATA/pg_tblspc/spcoid/PG_MAJORVER_CATVER/dboid/relfilenode
 * 例如：
 *			$PGDATA/pg_tblspc/20981/PG_9.0_201002161/719849/83292814
 *
 * 在 initdb 时创建了两个表空间：pg_global（用于共享表）和 pg_default（用于其他所有表）。为了向后兼容并保持在没有符号链接的平台上的功能，这些表空间的访问方式是特殊的：分别是
 *			$PGDATA/global/relfilenode
 *			$PGDATA/base/dboid/relfilenode
 *
 * 为了允许 CREATE DATABASE 为新数据库提供一个不同于模板数据库默认的默认表空间，我们规定 pg_class.reltablespace 中的零表示数据库的默认表空间。没有这个，CREATE DATABASE 将不得不进入并修改新数据库的系统目录。
 *
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 *
 * IDENTIFICATION
 *	  src/backend/commands/tablespace.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include <unistd.h>
#include <dirent.h>
#include <sys/stat.h>

#include "access/heapam.h"
#include "access/htup_details.h"
#include "access/reloptions.h"
#include "access/sysattr.h"
#include "access/tableam.h"
#include "access/xact.h"
#include "access/xloginsert.h"
#include "access/xlogutils.h"
#include "catalog/binary_upgrade.h"
#include "catalog/catalog.h"
#include "catalog/dependency.h"
#include "catalog/indexing.h"
#include "catalog/namespace.h"
#include "catalog/objectaccess.h"
#include "catalog/pg_namespace.h"
#include "catalog/pg_tablespace.h"
#include "commands/comment.h"
#include "commands/seclabel.h"
#include "commands/tablecmds.h"
#include "commands/tablespace.h"
#include "common/file_perm.h"
#include "miscadmin.h"
#include "postmaster/bgwriter.h"
#include "storage/fd.h"
#include "storage/lmgr.h"
#include "storage/standby.h"
#include "utils/acl.h"
#include "utils/builtins.h"
#include "utils/fmgroids.h"
#include "utils/guc.h"
#include "utils/lsyscache.h"
#include "utils/memutils.h"
#include "utils/rel.h"
#include "utils/varlena.h"

/* GUC 变量 */
char	   *default_tablespace = NULL;
char	   *temp_tablespaces = NULL;
bool		allow_in_place_tablespaces = false;

Oid			binary_upgrade_next_pg_tablespace_oid = InvalidOid;

static void fc_create_tablespace_directories(const char *fc_location,
										  const Oid fc_tablespaceoid);
static bool fc_destroy_tablespace_directories(Oid fc_tablespaceoid, bool fc_redo);


/*
 * 每个使用表空间的数据库通过一个以数据库 OID 命名的子目录被隔离到自己的命名空间中。
 * 在表空间中首次创建对象时，创建子目录。如果子目录已经存在，则静默跳过。
 *
 * isRedo 指示我们在 WAL 回放期间创建对象。
 * 在这种情况下，我们将处理表空间目录可能不存在的情况 --- 如果我们正在回放
 * 在一个随后被删除的表空间中的表上的操作，这可能会发生。
 * 我们通过在表空间符号链接正常存在的地方创建一个目录来处理这个问题。
 * 当然，这不是一个精确的回放，但根据可用信息，这是我们能做的最好事情。
 *
 * 如果不支持表空间，我们仍然需要它，以防我们在 WAL 回放期间
 * 需要重新创建数据库子目录（位于 $PGDATA/base 中）。
 */
void TablespaceCreateDbspace(Oid fc_spcNode, Oid fc_dbNode, bool fc_isRedo)
{
	struct stat fc_st;
	char	   *fc_dir;

	/*
	 * 全局表空间没有每个数据库的子目录，所以对此没有任何操作。
	 */
	if (fc_spcNode == GLOBALTABLESPACE_OID)
		return;

	Assert(OidIsValid(fc_spcNode));
	Assert(OidIsValid(fc_dbNode));

	fc_dir = GetDatabasePath(fc_dbNode, fc_spcNode);

	if (stat(fc_dir, &fc_st) < 0)
	{
		/* 目录不存在？ */
		if (errno == ENOENT)
		{
			/*
			 * 获取 TablespaceCreateLock，以确保没有同时进行的 DROP TABLESPACE
			 * 或 TablespaceCreateDbspace 操作。
			 */
			LWLockAcquire(TablespaceCreateLock, LW_EXCLUSIVE);

			/*
			 * 重新检查看看在我们等待锁的时候是否有人创建了目录。
			 */
			if (stat(fc_dir, &fc_st) == 0 && S_ISDIR(fc_st.st_mode))
			{
				/* 目录已创建 */
			}
			else
			{
				/* 目录创建失败？ */
				if (MakePGDirectory(fc_dir) < 0)
				{
					/* 除不存在或未在 WAL 回放中之外的失败？ */
					if (errno != ENOENT || !fc_isRedo)
						ereport(ERROR,
								(errcode_for_file_access(),
								 errmsg("could not create directory \"%s\": %m",
										fc_dir)));

					/*
					 * 在 WAL 回放期间，如果表空间在比我们当前正在回放的 WAL
					 * 流更前面被删除，几个级别的目录可能会缺失是可以想象的。
					 * 一种简单的解决方法是将它们作为普通目录创建，并希望
					 * 如果需要，它们会通过进一步的 WAL 回放被删除。
					 * 如果这也失败，我们就会遇到无法解决的问题，所以只需报告错误并退出。
					 */
					if (pg_mkdir_p(fc_dir, pg_dir_create_mode) < 0)
						ereport(ERROR,
								(errcode_for_file_access(),
								 errmsg("could not create directory \"%s\": %m",
										fc_dir)));
				}
			}

			LWLockRelease(TablespaceCreateLock);
		}
		else
		{
			ereport(ERROR,
					(errcode_for_file_access(),
					 errmsg("could not stat directory \"%s\": %m", fc_dir)));
		}
	}
	else
	{
		/* 这不是一个目录吗？ */
		if (!S_ISDIR(fc_st.st_mode))
			ereport(ERROR,
					(errcode(ERRCODE_WRONG_OBJECT_TYPE),
					 errmsg("\"%s\" exists but is not a directory",
							fc_dir)));
	}

	pfree(fc_dir);
}

/*
 * 创建一个表空间
 *
 * 只有超级用户才能创建表空间。这似乎是一个合理的限制，因为
 * 我们正在确定系统布局，此外，如果我们进行这种活动，可能已经是 root 用户。
 */
Oid CreateTableSpace(CreateTableSpaceStmt *fc_stmt)
{
#ifdef HAVE_SYMLINK
	Relation	fc_rel;
	Datum		fc_values[Natts_pg_tablespace];
	bool		fc_nulls[Natts_pg_tablespace];
	HeapTuple	fc_tuple;
	Oid			fc_tablespaceoid;
	char	   *fc_location;
	Oid			fc_ownerId;
	Datum		fc_newOptions;
	bool		fc_in_place;

	/* 必须是超级用户 */
	if (!superuser())
		ereport(ERROR,
				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
				 errmsg("permission denied to create tablespace \"%s\"",
						fc_stmt->tablespacename),
				 errhint("Must be superuser to create a tablespace.")));

	/* 但是，表空间的最终所有者不必是 */
	if (fc_stmt->owner)
		fc_ownerId = get_rolespec_oid(fc_stmt->owner, false);
	else
		fc_ownerId = GetUserId();

	/* Unix 化提供的路径，并去除任何尾随斜杠 */
	fc_location = pstrdup(fc_stmt->location);
	canonicalize_path(fc_location);

	/* 不允许使用引号，否则可能会危害 CREATE DATABASE */
	if (strchr(fc_location, '\''))
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_NAME),
				 errmsg("tablespace location cannot contain single quotes")));

	fc_in_place = allow_in_place_tablespaces && strlen(fc_location) == 0;

	/*
	 * 允许相对路径似乎风险很大
	 *
	 * 这也有助于确保位置不为空或全为空格，除非指定开发人员专用的就地表空间。
	 */
	if (!fc_in_place && !is_absolute_path(fc_location))
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
				 errmsg("tablespace location must be an absolute path")));

	/*
	 * 检查位置是否过长。请记住，我们将附加
	 * 'PG_XXX/<dboid>/<relid>_<fork>.<nnn>'。仅供参考，我们从未真正
	 * 引用整个路径，但 MakePGDirectory() 使用前两个部分。
	 */
	if (strlen(fc_location) + 1 + strlen(TABLESPACE_VERSION_DIRECTORY) + 1 +
		OIDCHARS + 1 + OIDCHARS + 1 + FORKNAMECHARS + 1 + OIDCHARS > MAXPGPATH)
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
				 errmsg("tablespace location \"%s\" is too long",
						fc_location)));

	/* 如果表空间在数据目录中，则发出警告。 */
	if (path_is_prefix_of_path(DataDir, fc_location))
		ereport(WARNING,
				(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
				 errmsg("tablespace location should not be inside the data directory")));

	/*
	 * 不允许创建名为 "pg_xxx" 的表空间；我们保留该命名空间用于系统目的。
	 */
	if (!allowSystemTableMods && IsReservedName(fc_stmt->tablespacename))
		ereport(ERROR,
				(errcode(ERRCODE_RESERVED_NAME),
				 errmsg("unacceptable tablespace name \"%s\"",
						fc_stmt->tablespacename),
				 errdetail("The prefix \"pg_\" is reserved for system tablespaces.")));

	/*
	 * 如果使用适当的开关构建，当违反表空间名称的回归测试约定时发出警告。
	 */
#ifdef ENFORCE_REGRESSION_TEST_NAME_RESTRICTIONS
	if (strncmp(fc_stmt->tablespacename, "regress_", 8) != 0)
		elog(WARNING, "tablespaces created by regression test cases should have names starting with \"regress_\"");
#endif

	/*
	 * 检查是否已经存在同名的表空间。（唯一索引无论如何会捕捉到这一点，但还是可以给出更友好的信息。）
	 */
	if (OidIsValid(get_tablespace_oid(fc_stmt->tablespacename, true)))
		ereport(ERROR,
				(errcode(ERRCODE_DUPLICATE_OBJECT),
				 errmsg("tablespace \"%s\" already exists",
						fc_stmt->tablespacename)));

	/*
	 * 将元组插入到 pg_tablespace 中。这样做的目的是为了锁定提议的表名，以防其它创建者。 如果我们在下面发现问题，插入将会回滚。
	 */
	fc_rel = table_open(TableSpaceRelationId, RowExclusiveLock);

	MemSet(fc_nulls, false, sizeof(fc_nulls));

	if (IsBinaryUpgrade)
	{
		/* 使用二进制升级覆盖表空间 oid */
		if (!OidIsValid(binary_upgrade_next_pg_tablespace_oid))
			ereport(ERROR,
					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
					 errmsg("pg_tablespace OID value not set when in binary upgrade mode")));

		fc_tablespaceoid = binary_upgrade_next_pg_tablespace_oid;
		binary_upgrade_next_pg_tablespace_oid = InvalidOid;
	}
	else
		fc_tablespaceoid = GetNewOidWithIndex(fc_rel, TablespaceOidIndexId,
										   Anum_pg_tablespace_oid);
	fc_values[Anum_pg_tablespace_oid - 1] = ObjectIdGetDatum(fc_tablespaceoid);
	fc_values[Anum_pg_tablespace_spcname - 1] =
		DirectFunctionCall1(namein, CStringGetDatum(fc_stmt->tablespacename));
	fc_values[Anum_pg_tablespace_spcowner - 1] =
		ObjectIdGetDatum(fc_ownerId);
	fc_nulls[Anum_pg_tablespace_spcacl - 1] = true;

	/* 生成新的提议 spcoptions（文本数组） */
	fc_newOptions = transformRelOptions((Datum) 0,
									 fc_stmt->options,
									 NULL, NULL, false, false);
	(void) tablespace_reloptions(fc_newOptions, true);
	if (fc_newOptions != (Datum) 0)
		fc_values[Anum_pg_tablespace_spcoptions - 1] = fc_newOptions;
	else
		fc_nulls[Anum_pg_tablespace_spcoptions - 1] = true;

	fc_tuple = heap_form_tuple(fc_rel->rd_att, fc_values, fc_nulls);

	CatalogTupleInsert(fc_rel, fc_tuple);

	heap_freetuple(fc_tuple);

	/* 记录对所有者的依赖性 */
	recordDependencyOnOwner(TableSpaceRelationId, fc_tablespaceoid, fc_ownerId);

	/* 新表空间的创建后钩子 */
	InvokeObjectPostCreateHook(TableSpaceRelationId, fc_tablespaceoid, 0);

	fc_create_tablespace_directories(fc_location, fc_tablespaceoid);

	/* 在 XLOG 中记录文件系统更改 */
	{
		xl_tblspc_create_rec fc_xlrec;

		fc_xlrec.ts_id = fc_tablespaceoid;

		XLogBeginInsert();
		XLogRegisterData((char *) &fc_xlrec,
						 offsetof(xl_tblspc_create_rec, ts_path));
		XLogRegisterData((char *) fc_location, strlen(fc_location) + 1);

		(void) XLogInsert(RM_TBLSPC_ID, XLOG_TBLSPC_CREATE);
	}

	/*
	 * 强制同步提交，以最小化创建磁盘上符号链接与标记事务提交之间的窗口。 任何窗口的存在都不好，但我们绝对不希望将其扩大到不必要的程度。
	 */
	ForceSyncCommit();

	pfree(fc_location);

	/* 我们在提交前保持对 pg_tablespace 的锁定 */
	table_close(fc_rel, NoLock);

	return fc_tablespaceoid;
#else							/* !HAVE_SYMLINK */
	ereport(ERROR,
			(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
			 errmsg("tablespaces are not supported on this platform")));
	return InvalidOid;			/* 保持编译器安静 */
#endif							/* HAVE_SYMLINK */
}

/*
 * 删除一个表空间
 *
 * 请注意检查表空间是否为空。
 */
void DropTableSpace(DropTableSpaceStmt *fc_stmt)
{
#ifdef HAVE_SYMLINK
	char	   *fc_tablespacename = fc_stmt->tablespacename;
	TableScanDesc fc_scandesc;
	Relation	fc_rel;
	HeapTuple	fc_tuple;
	Form_pg_tablespace fc_spcform;
	ScanKeyData fc_entry[1];
	Oid			fc_tablespaceoid;
	char	   *fc_detail;
	char	   *fc_detail_log;

	/*
	 * 查找目标元组
	 */
	fc_rel = table_open(TableSpaceRelationId, RowExclusiveLock);

	ScanKeyInit(&fc_entry[0],
				Anum_pg_tablespace_spcname,
				BTEqualStrategyNumber, F_NAMEEQ,
				CStringGetDatum(fc_tablespacename));
	fc_scandesc = table_beginscan_catalog(fc_rel, 1, fc_entry);
	fc_tuple = heap_getnext(fc_scandesc, ForwardScanDirection);

	if (!HeapTupleIsValid(fc_tuple))
	{
		if (!fc_stmt->missing_ok)
		{
			ereport(ERROR,
					(errcode(ERRCODE_UNDEFINED_OBJECT),
					 errmsg("tablespace \"%s\" does not exist",
							fc_tablespacename)));
		}
		else
		{
			ereport(NOTICE,
					(errmsg("tablespace \"%s\" does not exist, skipping",
							fc_tablespacename)));
			table_endscan(fc_scandesc);
			table_close(fc_rel, NoLock);
		}
		return;
	}

	fc_spcform = (Form_pg_tablespace) GETSTRUCT(fc_tuple);
	fc_tablespaceoid = fc_spcform->oid;

	/* 必须是表空间所有者 */
	if (!pg_tablespace_ownercheck(fc_tablespaceoid, GetUserId()))
		aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_TABLESPACE,
					   fc_tablespacename);

	/* 不允许删除标准表空间，即使是超级用户也不行 */
	if (IsPinnedObject(TableSpaceRelationId, fc_tablespaceoid))
		aclcheck_error(ACLCHECK_NO_PRIV, OBJECT_TABLESPACE,
					   fc_tablespacename);

	/* 检查依赖于此表空间的 pg_shdepend 条目 */
	if (checkSharedDependencies(TableSpaceRelationId, fc_tablespaceoid,
								&fc_detail, &fc_detail_log))
		ereport(ERROR,
				(errcode(ERRCODE_DEPENDENT_OBJECTS_STILL_EXIST),
				 errmsg("tablespace \"%s\" cannot be dropped because some objects depend on it",
						fc_tablespacename),
				 errdetail_internal("%s", fc_detail),
				 errdetail_log("%s", fc_detail_log)));

	/* 正在删除的表空间的 DROP 钩子 */
	InvokeObjectDropHook(TableSpaceRelationId, fc_tablespaceoid, 0);

	/*
	 * 移除 pg_tablespace 元组（如果我们下面失败，这将回滚）
	 */
	CatalogTupleDelete(fc_rel, &fc_tuple->t_self);

	table_endscan(fc_scandesc);

	/*
	 * 移除该表空间上的任何注释或安全标签。
	 */
	DeleteSharedComments(fc_tablespaceoid, TableSpaceRelationId);
	DeleteSharedSecurityLabel(fc_tablespaceoid, TableSpaceRelationId);

	/*
	 * 移除对所有者的依赖性。
	 */
	deleteSharedDependencyRecordsFor(TableSpaceRelationId, fc_tablespaceoid, 0);

	/*
	 * 获取 TablespaceCreateLock 以确保没有同时运行的 TablespaceCreateDbspace。
	 */
	LWLockAcquire(TablespaceCreateLock, LW_EXCLUSIVE);

	/*
	 * 尝试移除物理基础设施。
	 */
	if (!fc_destroy_tablespace_directories(fc_tablespaceoid, false))
	{
		/*
		 * 并非所有文件都已删除？但是，目录中可能会残留空文件，这些文件是由于例如 DROP TABLE 而留下的，已在下一个检查点计划删除（有关详细信息，请参见 mdunlink() 中的注释）。 我们可以立即将其删除，但我们无法将其与必须保留的重要数据文件区分开来。因此，我们强制进行检查点，这将清理任何残留文件，并重试。
		 */
		RequestCheckpoint(CHECKPOINT_IMMEDIATE | CHECKPOINT_FORCE | CHECKPOINT_WAIT);

		/*
		 * 在 Windows 上，未链接的文件会在目录列表中保留，直到没有进程保持对该文件的打开句柄。计划对文件进行 unlink 的 DDL 命令会发送失效消息，指示其他 PostgreSQL 进程关闭文件，但没有任何保证它们会及时处理它们。因此，我们还将使用全局屏障要求所有后端关闭所有文件，并等待它们完成。
		 */
		LWLockRelease(TablespaceCreateLock);
		WaitForProcSignalBarrier(EmitProcSignalBarrier(PROCSIGNAL_BARRIER_SMGRRELEASE));
		LWLockAcquire(TablespaceCreateLock, LW_EXCLUSIVE);

		/* 现在再试一次。 */
		if (!fc_destroy_tablespace_directories(fc_tablespaceoid, false))
		{
			/* 仍然不为空，文件一定很重要 */
			ereport(ERROR,
					(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
					 errmsg("tablespace \"%s\" is not empty",
							fc_tablespacename)));
		}
	}

	/* 在 XLOG 中记录文件系统更改 */
	{
		xl_tblspc_drop_rec fc_xlrec;

		fc_xlrec.ts_id = fc_tablespaceoid;

		XLogBeginInsert();
		XLogRegisterData((char *) &fc_xlrec, sizeof(xl_tblspc_drop_rec));

		(void) XLogInsert(RM_TBLSPC_ID, XLOG_TBLSPC_DROP);
	}

	/*
	 * 注意：因为我们检查了表空间是空的，所以不需要担心
	 * 刷新共享缓冲区或表空间中关系的空闲空间映射条目。
	 */

	/*
	 * 强制同步提交，以最小化在从磁盘上删除文件与标记事务已提交
	 * 之间的窗口。 任何窗口的存在都不是很好，但我们绝对不想
	 * 让它比必要的更大。
	 */
	ForceSyncCommit();

	/*
	 * 允许再次创建TablespaceCreateDbspace。
	 */
	LWLockRelease(TablespaceCreateLock);

	/* 我们在提交前保持对 pg_tablespace 的锁定 */
	table_close(fc_rel, NoLock);
#else							/* !HAVE_SYMLINK */
	ereport(ERROR,
			(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
			 errmsg("tablespaces are not supported on this platform")));
#endif							/* HAVE_SYMLINK */
}


/*
 * create_tablespace_directories
 *
 * 尝试创建文件系统基础设施，将$PGDATA/pg_tblspc/
 * 链接到指定目录
 */
static void fc_create_tablespace_directories(const char *fc_location, const Oid fc_tablespaceoid)
{
	char	   *fc_linkloc;
	char	   *fc_location_with_version_dir;
	struct stat fc_st;
	bool		fc_in_place;

	fc_linkloc = psprintf("pg_tblspc/%u", fc_tablespaceoid);

	/*
	 * 如果我们被要求制作一个“就地”表空间，则直接在符号链接
	 * 通常会去的地方创建目录。 目前这是一个仅限开发人员的
	 * 选项，以便于回归测试。
	 */
	fc_in_place = strlen(fc_location) == 0;

	if (fc_in_place)
	{
		if (MakePGDirectory(fc_linkloc) < 0 && errno != EEXIST)
			ereport(ERROR,
					(errcode_for_file_access(),
					 errmsg("could not create directory \"%s\": %m",
							fc_linkloc)));
	}

	fc_location_with_version_dir = psprintf("%s/%s", fc_in_place ? fc_linkloc : fc_location,
										 TABLESPACE_VERSION_DIRECTORY);

	/*
	 * 尝试将目标目录强制为安全权限。 如果这失败，
	 * 则该目录不存在或拥有者错误。 在就地模式下不需要，
	 * 因为在这种情况下我们是用所需的权限创建了目录。
	 */
	if (!fc_in_place && chmod(fc_location, pg_dir_create_mode) != 0)
	{
		if (errno == ENOENT)
			ereport(ERROR,
					(errcode(ERRCODE_UNDEFINED_FILE),
					 errmsg("directory \"%s\" does not exist", fc_location),
					 InRecovery ? errhint("Create this directory for the tablespace before "
										  "restarting the server.") : 0));
		else
			ereport(ERROR,
					(errcode_for_file_access(),
					 errmsg("could not set permissions on directory \"%s\": %m",
							fc_location)));
	}

	/*
	 * 版本目录的创建防止在单个位置有多个表空间。
	 * 这模仿了TablespaceCreateDbspace()，但它忽略了并发和缺失的父目录。
	 * 在没有父目录的情况下，chmod() 将失败。 pg_tablespace_spcname_index
	 * 防止并发。
	 */
	if (stat(fc_location_with_version_dir, &fc_st) < 0)
	{
		if (errno != ENOENT)
			ereport(ERROR,
					(errcode_for_file_access(),
					 errmsg("could not stat directory \"%s\": %m",
							fc_location_with_version_dir)));
		else if (MakePGDirectory(fc_location_with_version_dir) < 0)
			ereport(ERROR,
					(errcode_for_file_access(),
					 errmsg("could not create directory \"%s\": %m",
							fc_location_with_version_dir)));
	}
	else if (!S_ISDIR(fc_st.st_mode))
		ereport(ERROR,
				(errcode(ERRCODE_WRONG_OBJECT_TYPE),
				 errmsg("\"%s\" exists but is not a directory",
						fc_location_with_version_dir)));
	else if (!InRecovery)
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_IN_USE),
				 errmsg("directory \"%s\" already in use as a tablespace",
						fc_location_with_version_dir)));

	/*
	 * 在恢复过程中，移除旧的符号链接，以防它指向错误的地方。
	 */
	if (!fc_in_place && InRecovery)
		remove_tablespace_symlink(fc_linkloc);

	/*
	 * 在PGDATA下创建符号链接
	 */
	if (!fc_in_place && symlink(fc_location, fc_linkloc) < 0)
		ereport(ERROR,
				(errcode_for_file_access(),
				 errmsg("could not create symbolic link \"%s\": %m",
						fc_linkloc)));

	pfree(fc_linkloc);
	pfree(fc_location_with_version_dir);
}


/*
 * destroy_tablespace_directories
 *
 * 尝试移除表空间的文件系统基础设施。
 *
 * 'redo'表示我们正在从XLOG重做删除；在这种情况下，我们不应该
 * 因为问题而抛出错误，只需记录它们。 不删除这里的文件的最坏后果
 * 是未能释放一些磁盘空间，这并不值得抛出一个需要手动干预
 * 才能让数据库再次运行的错误。
 *
 * 成功则返回true，某个子目录不为空则返回false
 */
static bool fc_destroy_tablespace_directories(Oid fc_tablespaceoid, bool fc_redo)
{
	char	   *fc_linkloc;
	char	   *fc_linkloc_with_version_dir;
	DIR		   *fc_dirdesc;
	struct dirent *fc_de;
	char	   *fc_subfile;
	struct stat fc_st;

	fc_linkloc_with_version_dir = psprintf("pg_tblspc/%u/%s", fc_tablespaceoid,
										TABLESPACE_VERSION_DIRECTORY);

	/*
	 * 检查表空间是否还包含任何文件。我们尝试删除它中找到的每个数据库目录。rmdir 失败意味着该子目录中仍然存在文件，因此放弃。 (我们不必担心撤销任何已完成的 rmdirs，因为下次尝试从该数据库使用表空间时，将通过 TablespaceCreateDbspace 重新创建子目录。)
	 *
	 * 由于我们持有 TablespaceCreateLock，其他人不应同时创建任何新的子目录。然而，子目录中可能正在创建新的文件，因此 rmdir 调用可能会失败。最糟糕的后果是更友好的错误消息。
	 *
	 * 如果 redo 为真，则 ENOENT 很可能是此处的结果，我们允许它在没有评论的情况下通过。在正常操作中，我们仍然允许它，但会发出警告。这是因为即使 ProcessUtility 在事务块中不允许 DROP TABLESPACE，但以前的 DROP 可能已经失败并在删除表空间目录和/或符号链接后回滚。我们希望允许新的 DROP 尝试成功删除目录条目（以及如果仍存在的符号链接），因此我们不应在这里给出硬错误。
	 */
	fc_dirdesc = AllocateDir(fc_linkloc_with_version_dir);
	if (fc_dirdesc == NULL)
	{
		if (errno == ENOENT)
		{
			if (!fc_redo)
				ereport(WARNING,
						(errcode_for_file_access(),
						 errmsg("could not open directory \"%s\": %m",
								fc_linkloc_with_version_dir)));
			/* 符号链接可能仍然存在，因此去尝试删除它 */
			goto remove_symlink;
		}
		else if (fc_redo)
		{
			/* 在 redo 中，只记录其他类型的错误 */
			ereport(LOG,
					(errcode_for_file_access(),
					 errmsg("could not open directory \"%s\": %m",
							fc_linkloc_with_version_dir)));
			pfree(fc_linkloc_with_version_dir);
			return false;
		}
		/* 否则让 ReadDir 报告错误 */
	}

	while ((fc_de = ReadDir(fc_dirdesc, fc_linkloc_with_version_dir)) != NULL)
	{
		if (strcmp(fc_de->d_name, ".") == 0 ||
			strcmp(fc_de->d_name, "..") == 0)
			continue;

		fc_subfile = psprintf("%s/%s", fc_linkloc_with_version_dir, fc_de->d_name);

		/* 此检查仅是为了提供更友好的错误消息 */
		if (!fc_redo && !directory_is_empty(fc_subfile))
		{
			FreeDir(fc_dirdesc);
			pfree(fc_subfile);
			pfree(fc_linkloc_with_version_dir);
			return false;
		}

		/* 删除空目录 */
		if (rmdir(fc_subfile) < 0)
			ereport(fc_redo ? LOG : ERROR,
					(errcode_for_file_access(),
					 errmsg("could not remove directory \"%s\": %m",
							fc_subfile)));

		pfree(fc_subfile);
	}

	FreeDir(fc_dirdesc);

	/* 删除版本目录 */
	if (rmdir(fc_linkloc_with_version_dir) < 0)
	{
		ereport(fc_redo ? LOG : ERROR,
				(errcode_for_file_access(),
				 errmsg("could not remove directory \"%s\": %m",
						fc_linkloc_with_version_dir)));
		pfree(fc_linkloc_with_version_dir);
		return false;
	}

	/*
	 * 尝试删除符号链接。然而，我们必须处理它是目录而不是符号链接的可能性——这可能发生在 WAL 重放期间（见 TablespaceCreateDbspace）。
	 *
	 * 注意：在 redo 情况下，如果此最终步骤失败，我们将返回 true；重试没有意义。而且，ENOENT 不应引发超过警告的错误。
	 */
remove_symlink:
	fc_linkloc = pstrdup(fc_linkloc_with_version_dir);
	get_parent_directory(fc_linkloc);
	if (lstat(fc_linkloc, &fc_st) < 0)
	{
		int			fc_saved_errno = errno;

		ereport(fc_redo ? LOG : (fc_saved_errno == ENOENT ? WARNING : ERROR),
				(errcode_for_file_access(),
				 errmsg("could not stat file \"%s\": %m",
						fc_linkloc)));
	}
	else if (S_ISDIR(fc_st.st_mode))
	{
		if (rmdir(fc_linkloc) < 0)
		{
			int			fc_saved_errno = errno;

			ereport(fc_redo ? LOG : (fc_saved_errno == ENOENT ? WARNING : ERROR),
					(errcode_for_file_access(),
					 errmsg("could not remove directory \"%s\": %m",
							fc_linkloc)));
		}
	}
	else if (S_ISLNK(fc_st.st_mode))
	{
		if (unlink(fc_linkloc) < 0)
		{
			int			fc_saved_errno = errno;

			ereport(fc_redo ? LOG : (fc_saved_errno == ENOENT ? WARNING : ERROR),
					(errcode_for_file_access(),
					 errmsg("could not remove symbolic link \"%s\": %m",
							fc_linkloc)));
		}
	}
	else
	{
		/* 拒绝删除任何不是目录或符号链接的内容 */
		ereport(fc_redo ? LOG : ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("\"%s\" is not a directory or symbolic link",
						fc_linkloc)));
	}

	pfree(fc_linkloc_with_version_dir);
	pfree(fc_linkloc);

	return true;
}


/*
 * 检查目录是否为空。
 *
 * 这可能应该放在其他地方，但不确定在哪里……
 */
bool directory_is_empty(const char *fc_path)
{
	DIR		   *fc_dirdesc;
	struct dirent *fc_de;

	fc_dirdesc = AllocateDir(fc_path);

	while ((fc_de = ReadDir(fc_dirdesc, fc_path)) != NULL)
	{
		if (strcmp(fc_de->d_name, ".") == 0 ||
			strcmp(fc_de->d_name, "..") == 0)
			continue;
		FreeDir(fc_dirdesc);
		return false;
	}

	FreeDir(fc_dirdesc);
	return true;
}

/*
 *	remove_tablespace_symlink
 *
 * 此函数删除 pg_tblspc 中的符号链接。在 Windows 上，连接点表现得像目录，因此我们必须能够应用 rmdir。此函数的工作方式类似于 destroy_tablespace_directories 中的符号链接删除代码，不同之处在于删除失败始终是一个错误。但是，如果文件根本不存在，那是可以的。
 */
void remove_tablespace_symlink(const char *fc_linkloc)
{
	struct stat fc_st;

	if (lstat(fc_linkloc, &fc_st) < 0)
	{
		if (errno == ENOENT)
			return;
		ereport(ERROR,
				(errcode_for_file_access(),
				 errmsg("could not stat file \"%s\": %m", fc_linkloc)));
	}

	if (S_ISDIR(fc_st.st_mode))
	{
		/*
		 * 如果目录不为空，则此操作将失败，但如果是连接点则不会。
		 */
		if (rmdir(fc_linkloc) < 0 && errno != ENOENT)
			ereport(ERROR,
					(errcode_for_file_access(),
					 errmsg("could not remove directory \"%s\": %m",
							fc_linkloc)));
	}
	else if (S_ISLNK(fc_st.st_mode))
	{
		if (unlink(fc_linkloc) < 0 && errno != ENOENT)
			ereport(ERROR,
					(errcode_for_file_access(),
					 errmsg("could not remove symbolic link \"%s\": %m",
							fc_linkloc)));
	}
	else
	{
		/* 拒绝删除任何不是目录或符号链接的内容 */
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("\"%s\" is not a directory or symbolic link",
						fc_linkloc)));
	}
}

/*
 * 重命名表空间
 */
ObjectAddress RenameTableSpace(const char *fc_oldname, const char *fc_newname)
{
	Oid			fc_tspId;
	Relation	fc_rel;
	ScanKeyData fc_entry[1];
	TableScanDesc fc_scan;
	HeapTuple	fc_tup;
	HeapTuple	fc_newtuple;
	Form_pg_tablespace fc_newform;
	ObjectAddress fc_address;

	/* 搜索 pg_tablespace */
	fc_rel = table_open(TableSpaceRelationId, RowExclusiveLock);

	ScanKeyInit(&fc_entry[0],
				Anum_pg_tablespace_spcname,
				BTEqualStrategyNumber, F_NAMEEQ,
				CStringGetDatum(fc_oldname));
	fc_scan = table_beginscan_catalog(fc_rel, 1, fc_entry);
	fc_tup = heap_getnext(fc_scan, ForwardScanDirection);
	if (!HeapTupleIsValid(fc_tup))
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_OBJECT),
				 errmsg("tablespace \"%s\" does not exist",
						fc_oldname)));

	fc_newtuple = heap_copytuple(fc_tup);
	fc_newform = (Form_pg_tablespace) GETSTRUCT(fc_newtuple);
	fc_tspId = fc_newform->oid;

	table_endscan(fc_scan);

	/* 必须是拥有者 */
	if (!pg_tablespace_ownercheck(fc_tspId, GetUserId()))
		aclcheck_error(ACLCHECK_NO_PRIV, OBJECT_TABLESPACE, fc_oldname);

	/* 验证新名称 */
	if (!allowSystemTableMods && IsReservedName(fc_newname))
		ereport(ERROR,
				(errcode(ERRCODE_RESERVED_NAME),
				 errmsg("unacceptable tablespace name \"%s\"", fc_newname),
				 errdetail("The prefix \"pg_\" is reserved for system tablespaces.")));

	/*
	 * 如果使用适当的开关构建，当违反表空间名称的回归测试约定时发出警告。
	 */
#ifdef ENFORCE_REGRESSION_TEST_NAME_RESTRICTIONS
	if (strncmp(fc_newname, "regress_", 8) != 0)
		elog(WARNING, "tablespaces created by regression test cases should have names starting with \"regress_\"");
#endif

	/* 确保新名称不存在 */
	ScanKeyInit(&fc_entry[0],
				Anum_pg_tablespace_spcname,
				BTEqualStrategyNumber, F_NAMEEQ,
				CStringGetDatum(fc_newname));
	fc_scan = table_beginscan_catalog(fc_rel, 1, fc_entry);
	fc_tup = heap_getnext(fc_scan, ForwardScanDirection);
	if (HeapTupleIsValid(fc_tup))
		ereport(ERROR,
				(errcode(ERRCODE_DUPLICATE_OBJECT),
				 errmsg("tablespace \"%s\" already exists",
						fc_newname)));

	table_endscan(fc_scan);

	/* 好的，更新条目 */
	namestrcpy(&(fc_newform->spcname), fc_newname);

	CatalogTupleUpdate(fc_rel, &fc_newtuple->t_self, fc_newtuple);

	InvokeObjectPostAlterHook(TableSpaceRelationId, fc_tspId, 0);

	ObjectAddressSet(fc_address, TableSpaceRelationId, fc_tspId);

	table_close(fc_rel, NoLock);

	return fc_address;
}

/*
 * 更改表空间选项
 */
Oid AlterTableSpaceOptions(AlterTableSpaceOptionsStmt *fc_stmt)
{
	Relation	fc_rel;
	ScanKeyData fc_entry[1];
	TableScanDesc fc_scandesc;
	HeapTuple	fc_tup;
	Oid			fc_tablespaceoid;
	Datum		fc_datum;
	Datum		fc_newOptions;
	Datum		fc_repl_val[Natts_pg_tablespace];
	bool		fc_isnull;
	bool		fc_repl_null[Natts_pg_tablespace];
	bool		fc_repl_repl[Natts_pg_tablespace];
	HeapTuple	fc_newtuple;

	/* 搜索 pg_tablespace */
	fc_rel = table_open(TableSpaceRelationId, RowExclusiveLock);

	ScanKeyInit(&fc_entry[0],
				Anum_pg_tablespace_spcname,
				BTEqualStrategyNumber, F_NAMEEQ,
				CStringGetDatum(fc_stmt->tablespacename));
	fc_scandesc = table_beginscan_catalog(fc_rel, 1, fc_entry);
	fc_tup = heap_getnext(fc_scandesc, ForwardScanDirection);
	if (!HeapTupleIsValid(fc_tup))
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_OBJECT),
				 errmsg("tablespace \"%s\" does not exist",
						fc_stmt->tablespacename)));

	fc_tablespaceoid = ((Form_pg_tablespace) GETSTRUCT(fc_tup))->oid;

	/* 必须是现有对象的所有者 */
	if (!pg_tablespace_ownercheck(fc_tablespaceoid, GetUserId()))
		aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_TABLESPACE,
					   fc_stmt->tablespacename);

	/* 生成新的提议 spcoptions（文本数组） */
	fc_datum = heap_getattr(fc_tup, Anum_pg_tablespace_spcoptions,
						 RelationGetDescr(fc_rel), &fc_isnull);
	fc_newOptions = transformRelOptions(fc_isnull ? (Datum) 0 : fc_datum,
									 fc_stmt->options, NULL, NULL, false,
									 fc_stmt->isReset);
	(void) tablespace_reloptions(fc_newOptions, true);

	/* 构建新元组。 */
	memset(fc_repl_null, false, sizeof(fc_repl_null));
	memset(fc_repl_repl, false, sizeof(fc_repl_repl));
	if (fc_newOptions != (Datum) 0)
		fc_repl_val[Anum_pg_tablespace_spcoptions - 1] = fc_newOptions;
	else
		fc_repl_null[Anum_pg_tablespace_spcoptions - 1] = true;
	fc_repl_repl[Anum_pg_tablespace_spcoptions - 1] = true;
	fc_newtuple = heap_modify_tuple(fc_tup, RelationGetDescr(fc_rel), fc_repl_val,
								 fc_repl_null, fc_repl_repl);

	/* 更新系统目录。 */
	CatalogTupleUpdate(fc_rel, &fc_newtuple->t_self, fc_newtuple);

	InvokeObjectPostAlterHook(TableSpaceRelationId, fc_tablespaceoid, 0);

	heap_freetuple(fc_newtuple);

	/* 结束堆扫描。 */
	table_endscan(fc_scandesc);
	table_close(fc_rel, NoLock);

	return fc_tablespaceoid;
}

/*
 * 处理 GUC 变量 'default_tablespace' 的例程。
 */

/* check_hook: 验证新的 default_tablespace */
bool check_default_tablespace(char **fc_newval, void **fc_extra, GucSource fc_source)
{
	/*
	 * 如果我们不在事务中，或未连接到数据库，
	 * 就无法进行验证名称所需的目录访问。必须
	 * 信任该值。
	 */
	if (IsTransactionState() && MyDatabaseId != InvalidOid)
	{
		if (**fc_newval != '\0' &&
			!OidIsValid(get_tablespace_oid(*fc_newval, true)))
		{
			/*
			 * 当 source == PGC_S_TEST 时，对于不存在的
			 * 表空间不要抛出硬错误，仅发出 NOTICE。参见 guc.h 中的注释。
			 */
			if (fc_source == PGC_S_TEST)
			{
				ereport(NOTICE,
						(errcode(ERRCODE_UNDEFINED_OBJECT),
						 errmsg("tablespace \"%s\" does not exist",
								*fc_newval)));
			}
			else
			{
				GUC_check_errdetail("Tablespace \"%s\" does not exist.",
									*fc_newval);
				return false;
			}
		}
	}

	return true;
}

/*
 * GetDefaultTablespace -- 获取当前默认表空间的 OID
 *
 * 临时对象有不同的默认表空间，因此必须指定
 * relpersistence 参数。此外，对于分区表，
 * 我们不允许指定数据库默认值，因此也需要指定。
 *
 * 可能返回 InvalidOid 表示“使用数据库的默认表空间”。
 *
 * 注意，调用者应检查任何
 * 除 InvalidOid 之外的结果的适当权限。
 *
 * 这个存在是为了隐藏（并可能优化使用）
 * default_tablespace GUC 变量。
 */
Oid GetDefaultTablespace(char fc_relpersistence, bool fc_partitioned)
{
	Oid			fc_result;

	/* 临时表的情况在其他地方处理 */
	if (fc_relpersistence == RELPERSISTENCE_TEMP)
	{
		PrepareTempTablespaces();
		return GetNextTempTableSpace();
	}

	/* 默认表空间 == "" 的快速路径 */
	if (default_tablespace == NULL || default_tablespace[0] == '\0')
		return InvalidOid;

	/*
	 * 使用缓存查找以提高速度是很诱人的，但那样我们将
	 * 无法检测到表空间在设置 GUC 变量后被删除的情况。
	 * 还要注意，如果值不能引用现有的表空间，我们不会抱怨；
	 * 我们只是默默返回 InvalidOid，
	 * 导致新对象在数据库的表空间中创建。
	 */
	fc_result = get_tablespace_oid(default_tablespace, true);

	/*
	 * 允许在 default_tablespace 中明确指定数据库的默认表空间，
	 * 而不触发权限检查。但在创建分区表时不允许这样做，
	 * 因为结果会令人困惑。
	 */
	if (fc_result == MyDatabaseTableSpace)
	{
		if (fc_partitioned)
			ereport(ERROR,
					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
					 errmsg("cannot specify default tablespace for partitioned relations")));
		fc_result = InvalidOid;
	}
	return fc_result;
}


/*
 * 处理 GUC 变量 'temp_tablespaces' 的例程。
 */

typedef struct
{
	/* 要传递给 SetTempTablespaces() 的 OID 数组 */
	int			numSpcs;
	Oid			tblSpcs[FLEXIBLE_ARRAY_MEMBER];
} temp_tablespaces_extra;

/* check_hook: 验证新的 temp_tablespaces */
bool check_temp_tablespaces(char **fc_newval, void **fc_extra, GucSource fc_source)
{
	char	   *fc_rawname;
	List	   *fc_namelist;

	/* 需要可修改的字符串副本 */
	fc_rawname = pstrdup(*fc_newval);

	/* 将字符串解析为标识符列表 */
	if (!SplitIdentifierString(fc_rawname, ',', &fc_namelist))
	{
		/* 名称列表中的语法错误 */
		GUC_check_errdetail("List syntax is invalid.");
		pfree(fc_rawname);
		list_free(fc_namelist);
		return false;
	}

	/*
	 * 如果我们不在事务中，或未连接到数据库，
	 * 就无法进行验证名称所需的目录访问。必须
	 * 信任该值。幸运的是，这样也不需要
	 * 将数据传递给 fd.c。
	 */
	if (IsTransactionState() && MyDatabaseId != InvalidOid)
	{
		temp_tablespaces_extra *fc_myextra;
		Oid		   *fc_tblSpcs;
		int			fc_numSpcs;
		ListCell   *fc_l;

		/* 临时工作区，直到我们完成验证列表 */
		fc_tblSpcs = (Oid *) palloc(list_length(fc_namelist) * sizeof(Oid));
		fc_numSpcs = 0;
		foreach(fc_l, fc_namelist)
		{
			char	   *fc_curname = (char *) lfirst(fc_l);
			Oid			fc_curoid;
			AclResult	fc_aclresult;

			/* 允许空字符串（表示数据库默认值） */
			if (fc_curname[0] == '\0')
			{
				/* InvalidOid 表示数据库的默认表空间 */
				fc_tblSpcs[fc_numSpcs++] = InvalidOid;
				continue;
			}

			/*
			 * 在交互式 SET 命令中，我们报告错误的信息。当
			 * source == PGC_S_TEST 时，不会对不存在的表空间抛出硬错误，只给出提示。请参阅 guc.h 中的注释。
			 */
			fc_curoid = get_tablespace_oid(fc_curname, fc_source <= PGC_S_TEST);
			if (fc_curoid == InvalidOid)
			{
				if (fc_source == PGC_S_TEST)
					ereport(NOTICE,
							(errcode(ERRCODE_UNDEFINED_OBJECT),
							 errmsg("tablespace \"%s\" does not exist",
									fc_curname)));
				continue;
			}

			/*
			 * 允许在 temp_tablespaces 中显式指定数据库的默认表空间，而不触发权限检查。
			 */
			if (fc_curoid == MyDatabaseTableSpace)
			{
				/* InvalidOid 表示数据库的默认表空间 */
				fc_tblSpcs[fc_numSpcs++] = InvalidOid;
				continue;
			}

			/* 检查权限，类似地仅在交互时抱怨 */
			fc_aclresult = pg_tablespace_aclcheck(fc_curoid, GetUserId(),
											   ACL_CREATE);
			if (fc_aclresult != ACLCHECK_OK)
			{
				if (fc_source >= PGC_S_INTERACTIVE)
					aclcheck_error(fc_aclresult, OBJECT_TABLESPACE, fc_curname);
				continue;
			}

			fc_tblSpcs[fc_numSpcs++] = fc_curoid;
		}

		/* 现在为 assign_temp_tablespaces 准备一个“额外”的结构 */
		fc_myextra = malloc(offsetof(temp_tablespaces_extra, tblSpcs) +
						 fc_numSpcs * sizeof(Oid));
		if (!fc_myextra)
			return false;
		fc_myextra->numSpcs = fc_numSpcs;
		memcpy(fc_myextra->tblSpcs, fc_tblSpcs, fc_numSpcs * sizeof(Oid));
		*fc_extra = (void *) fc_myextra;

		pfree(fc_tblSpcs);
	}

	pfree(fc_rawname);
	list_free(fc_namelist);

	return true;
}

/* assign_hook：根据需要执行额外操作 */
void assign_temp_tablespaces(const char *fc_newval, void *fc_extra)
{
	temp_tablespaces_extra *fc_myextra = (temp_tablespaces_extra *) fc_extra;

	/*
	 * 如果 check_temp_tablespaces 在事务内执行，则将
	 * 生成的列表传递给 fd.c。否则，清除 fd.c 的列表；我们必须仍然处于事务外，或者在事务退出时恢复，
	 * 在这两种情况下，我们只需让下一个 PrepareTempTablespaces 调用
	 * 使事情正常。
	 */
	if (fc_myextra)
		SetTempTablespaces(fc_myextra->tblSpcs, fc_myextra->numSpcs);
	else
		SetTempTablespaces(NULL, 0);
}

/*
 * PrepareTempTablespaces -- 准备用于临时表空间
 *
 * 如果我们尚未在当前事务中执行该操作，则解析
 * temp_tablespaces GUC 变量并告诉 fd.c 要使用
 * 哪个表空间作为临时文件。
 */
void PrepareTempTablespaces(void)
{
	char	   *fc_rawname;
	List	   *fc_namelist;
	Oid		   *fc_tblSpcs;
	int			fc_numSpcs;
	ListCell   *fc_l;

	/* 如果在当前事务中已完成，则不执行任何操作 */
	if (TempTablespacesAreSet())
		return;

	/*
	 * 除非在事务内，否则无法进行目录访问。这只是一个
	 * 安全检查，以防此函数被低级代码调用，该代码可能在
	 * 事务外执行。请注意，在这种情况下，fd.c 将回退到使用
	 * 当前数据库的默认表空间，这通常是可以的。
	 */
	if (!IsTransactionState())
		return;

	/* 需要可修改的字符串副本 */
	fc_rawname = pstrdup(temp_tablespaces);

	/* 将字符串解析为标识符列表 */
	if (!SplitIdentifierString(fc_rawname, ',', &fc_namelist))
	{
		/* 名称列表中的语法错误 */
		SetTempTablespaces(NULL, 0);
		pfree(fc_rawname);
		list_free(fc_namelist);
		return;
	}

	/* 将表空间 OID 存储在 TopTransactionContext 的数组中 */
	fc_tblSpcs = (Oid *) MemoryContextAlloc(TopTransactionContext,
										 list_length(fc_namelist) * sizeof(Oid));
	fc_numSpcs = 0;
	foreach(fc_l, fc_namelist)
	{
		char	   *fc_curname = (char *) lfirst(fc_l);
		Oid			fc_curoid;
		AclResult	fc_aclresult;

		/* 允许空字符串（表示数据库默认值） */
		if (fc_curname[0] == '\0')
		{
			/* InvalidOid 表示数据库的默认表空间 */
			fc_tblSpcs[fc_numSpcs++] = InvalidOid;
			continue;
		}

		/* 否则验证名称是否为有效的表空间名称 */
		fc_curoid = get_tablespace_oid(fc_curname, true);
		if (fc_curoid == InvalidOid)
		{
			/* 跳过任何坏的列表元素 */
			continue;
		}

		/*
		 * 允许在 temp_tablespaces 中显式指定数据库的默认表空间而不触发权限检查。
		 */
		if (fc_curoid == MyDatabaseTableSpace)
		{
			/* InvalidOid 表示数据库的默认表空间 */
			fc_tblSpcs[fc_numSpcs++] = InvalidOid;
			continue;
		}

		/* 检查权限类似地 */
		fc_aclresult = pg_tablespace_aclcheck(fc_curoid, GetUserId(),
										   ACL_CREATE);
		if (fc_aclresult != ACLCHECK_OK)
			continue;

		fc_tblSpcs[fc_numSpcs++] = fc_curoid;
	}

	SetTempTablespaces(fc_tblSpcs, fc_numSpcs);

	pfree(fc_rawname);
	list_free(fc_namelist);
}


/*
 * get_tablespace_oid - 给定表空间名称，查找 OID
 *
 * 如果 missing_ok 为 false，则在未找到表空间名称时抛出错误。如果
 * 为 true，仅返回 InvalidOid。
 */
Oid get_tablespace_oid(const char *fc_tablespacename, bool fc_missing_ok)
{
	Oid			fc_result;
	Relation	fc_rel;
	TableScanDesc fc_scandesc;
	HeapTuple	fc_tuple;
	ScanKeyData fc_entry[1];

	/*
	 * 搜索 pg_tablespace。我们在这里使用堆扫描，即使名称上有索引，基于这样的理论：pg_tablespace 通常只有
	 * 少数条目，因此索引查找是浪费精力。
	 */
	fc_rel = table_open(TableSpaceRelationId, AccessShareLock);

	ScanKeyInit(&fc_entry[0],
				Anum_pg_tablespace_spcname,
				BTEqualStrategyNumber, F_NAMEEQ,
				CStringGetDatum(fc_tablespacename));
	fc_scandesc = table_beginscan_catalog(fc_rel, 1, fc_entry);
	fc_tuple = heap_getnext(fc_scandesc, ForwardScanDirection);

	/* 我们假设最多只有一个匹配的元组 */
	if (HeapTupleIsValid(fc_tuple))
		fc_result = ((Form_pg_tablespace) GETSTRUCT(fc_tuple))->oid;
	else
		fc_result = InvalidOid;

	table_endscan(fc_scandesc);
	table_close(fc_rel, AccessShareLock);

	if (!OidIsValid(fc_result) && !fc_missing_ok)
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_OBJECT),
				 errmsg("tablespace \"%s\" does not exist",
						fc_tablespacename)));

	return fc_result;
}

/*
 * get_tablespace_name - 给定表空间 OID，查找名称
 *
 * 返回一个 palloc 的字符串，如果没有这样的表空间则返回 NULL。
 */
char * get_tablespace_name(Oid fc_spc_oid)
{
	char	   *fc_result;
	Relation	fc_rel;
	TableScanDesc fc_scandesc;
	HeapTuple	fc_tuple;
	ScanKeyData fc_entry[1];

	/*
	 * 搜索 pg_tablespace。我们在这里使用堆扫描，即使 oid 上有索引，基于以下理论：pg_tablespace 通常只会有很少的条目，因此索引查找是浪费精力。
	 */
	fc_rel = table_open(TableSpaceRelationId, AccessShareLock);

	ScanKeyInit(&fc_entry[0],
				Anum_pg_tablespace_oid,
				BTEqualStrategyNumber, F_OIDEQ,
				ObjectIdGetDatum(fc_spc_oid));
	fc_scandesc = table_beginscan_catalog(fc_rel, 1, fc_entry);
	fc_tuple = heap_getnext(fc_scandesc, ForwardScanDirection);

	/* 我们假设最多只有一个匹配的元组 */
	if (HeapTupleIsValid(fc_tuple))
		fc_result = pstrdup(NameStr(((Form_pg_tablespace) GETSTRUCT(fc_tuple))->spcname));
	else
		fc_result = NULL;

	table_endscan(fc_scandesc);
	table_close(fc_rel, AccessShareLock);

	return fc_result;
}


/*
 * TABLESPACE 资源管理器的例程
 */
void tblspc_redo(XLogReaderState *fc_record)
{
	uint8		fc_info = XLogRecGetInfo(fc_record) & ~XLR_INFO_MASK;

	/* 在 tblspc 记录中不使用备份块 */
	Assert(!XLogRecHasAnyBlockRefs(fc_record));

	if (fc_info == XLOG_TBLSPC_CREATE)
	{
		xl_tblspc_create_rec *fc_xlrec = (xl_tblspc_create_rec *) XLogRecGetData(fc_record);
		char	   *fc_location = fc_xlrec->ts_path;

		fc_create_tablespace_directories(fc_location, fc_xlrec->ts_id);
	}
	else if (fc_info == XLOG_TBLSPC_DROP)
	{
		xl_tblspc_drop_rec *fc_xlrec = (xl_tblspc_drop_rec *) XLogRecGetData(fc_record);

		/* 关闭所有后端中的 smgr fds。 */
		WaitForProcSignalBarrier(EmitProcSignalBarrier(PROCSIGNAL_BARRIER_SMGRRELEASE));

		/*
		 * 如果我们为删除 tablespace 发出了 WAL 记录，这意味着
		 * 在删除时它内没有任何文件。这意味着在此时不能存在
		 * 永久对象。
		 *
		 * 备用用户可能正在使用此 tablespace 作为其临时文件的位置，
		 * 因此如果我们未能删除所有文件，则执行冲突处理并重试，
		 * 如果当前已启用。
		 *
		 * 失败的其他可能原因包括备用服务器上的文件权限问题，
		 * 而在主服务器上则没有问题，等等。对此我们无能为力，
		 * 所以只能删除我们能删除的部分并继续。
		 */
		if (!fc_destroy_tablespace_directories(fc_xlrec->ts_id, true))
		{
			ResolveRecoveryConflictWithTablespace(fc_xlrec->ts_id);

			/*
			 * 如果我们进行了恢复处理，那么希望写入临时文件的后端
			 * 此时应该已经清理并退出。因此在抱怨之前重试。
			 * 如果我们再次失败，这只是一个日志条件，
			 * 因为不值得抛出错误（因为这会导致数据库崩溃，
			 * 并需要手动干预才能在重启时超过此 WAL 记录）。
			 */
			if (!fc_destroy_tablespace_directories(fc_xlrec->ts_id, true))
				ereport(LOG,
						(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
						 errmsg("directories for tablespace %u could not be removed",
								fc_xlrec->ts_id),
						 errhint("You can remove the directories manually if necessary.")));
		}
	}
	else
		elog(PANIC, "tblspc_redo: unknown op code %u", fc_info);
}
